import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaPairRDD;
import org.apache.spark.api.java.JavaSparkContext;
import scala.Tuple2;

import java.util.Arrays;

/**
 *  读取hdfs文件，在yarn运行spark, 打印到文件里，
 */
public class WordCount {
    public static void main(String[] args) {
        SparkConf conf = new SparkConf()
                .setAppName("WordCount")
                .setMaster("yarn");// 使用 hadoop yarn 运行
        JavaSparkContext sc = new JavaSparkContext(conf);

        JavaRDD<String> lines = sc.textFile("hdfs://namenode:9000/text.log"); //读取hdfs根路径下的text.log文件
        JavaRDD<String> words = lines.flatMap(s -> Arrays.asList(s.split(" ")).iterator());
        JavaPairRDD<String, Integer> wordPairs = words.mapToPair(word -> new Tuple2<>(word, 1));
        JavaPairRDD<String, Integer> counts = wordPairs.reduceByKey((a, b) -> a + b);

        counts.saveAsTextFile("hdfs://namenode:9000/output/wordcount");
        sc.close();
    }
}